package hn.cch.hadoop.util;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class HdfsUtil {


    public static FileSystem connect(String hdfsUrl) throws IOException {
        Properties properties = System.getProperties();
        // 设置环境变量指定hadoop用户名解决权限问题（默认以主机名登录）
        properties.setProperty("HADOOP_USER_NAME", "root");
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", hdfsUrl);
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        return fileSystem;
    }


    public static List<Path> getAllFiles(FileSystem fileSystem, Path rootPath) throws IOException {
        // 第二个参数表示是否递归变量
        RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(rootPath, true);
        List<Path> pathList = new ArrayList<>();
        while (listFiles.hasNext()) {
            Path path = listFiles.next().getPath();
            pathList.add(path);
        }
        return pathList;
    }

    public static void download(FileSystem fileSystem, Path path, File file) throws IOException {
        FSDataInputStream fsDataInputStream = fileSystem.open(path);
        FileOutputStream fileOutputStream = new FileOutputStream(file);
        IOUtils.copy(fsDataInputStream, fileOutputStream);
        IOUtils.closeQuietly(fsDataInputStream);
        IOUtils.closeQuietly(fileOutputStream);
    }


    public static void upload(FileSystem fileSystem, Path filePath, Path hdfsPath) throws IOException {
        // 先判断文件夹是否存在，不存在就创建
        Path dirsPath = hdfsPath.getParent();
        if (!fileSystem.exists(dirsPath))
        {
            fileSystem.mkdirs(dirsPath);
        }
        fileSystem.copyFromLocalFile(filePath, hdfsPath);
    }


}
